use std::sync::Arc;
use shq_common::prelude::{ get_ch_pool, get_my_pool, get_named_lock_manager };
use sqlx::Row;
use tokio::sync::Mutex;

#[derive(Debug)]
pub struct KpiImportConfName {
    pub conf_name: String,
    pub zbid: i32,
}

async fn get_lock() -> Arc<Mutex<()>> {
    let lock_manager = get_named_lock_manager().await;
    lock_manager.get_lock("#kpi_import_lock").await
}

pub async fn kpi_import_conf_name(
    jd: i32,
    conf_name: &str
) -> anyhow::Result<Vec<KpiImportConfName>> {
    let sql =
        "SELECT ISDF, ZBID FROM pas_oltp_sh.SH_CONF_ZB_DEPT WHERE NF = LEFT(?, 4) AND (ZBMC1 = ? OR ZBMC2 = ?)";
    let rows = sqlx
    ::query(sql)
        .bind(jd.to_string())
        .bind(conf_name)
        .bind(conf_name)
        .fetch_all(get_my_pool()).await?;
    let mut vals = Vec::new();
    for row in rows.into_iter() {
        let perm_id: i32 = row.try_get(0)?;
        let zb_id: i32 = row.try_get(1)?;
        let conf_name = match perm_id {
            0 => "kpi_import_lhzb".to_string(),
            1 => "kpi_import_pjzb".to_string(),
            _ => {
                continue;
            }
        };
        vals.push(KpiImportConfName {
            conf_name: conf_name.clone(),
            zbid: zb_id,
        });
        install_kpi_import_log(zb_id, jd).await?;
        if conf_name == "kpi_import_lhzb" {
            let sql = "DELETE  from pas_oltp_sh.kpi_import_lhzb where jd = ? and zbid = ?";
            sqlx::query(sql).bind(jd).bind(zb_id).execute(get_my_pool()).await?;
        } else if conf_name == "kpi_import_pjzb" {
            let sql = "DELETE  from pas_oltp_sh.kpi_import_pjzb where jd = ? and zbid = ?";
            sqlx::query(sql).bind(jd).bind(zb_id).execute(get_my_pool()).await?;
        }
        let mut client = get_ch_pool().get_handle().await?;
        let drop_table_sql = format!("DROP TABLE IF EXISTS {}_{}_{}", conf_name, jd, zb_id);
        client.execute(&drop_table_sql).await?;
        let sql = "SELECT name FROM `system`.tables WHERE database = 'pas_olap_sh' AND (name like 'tmp_khbb_zhjyjh%' or name  like 'TMP_KHBB_ZHJYJH%')";
        let block = client.query(sql).fetch_all().await?;
        for row in block.rows() {
            let table_name: &str = row.get(0)?;
            let drop_table_sql = format!("DROP TABLE IF EXISTS {}", table_name);
            client.execute(&drop_table_sql).await?;
        }
    }
    Ok(vals)
}

pub async fn install_kpi_import_log(zbid: i32, jd: i32) -> anyhow::Result<()> {
    install_kpi_export_log(jd).await?;
    let sql = "SELECT COUNT(*)count  from pas_oltp_sh.kpi_import_log where jd = ? and zbid = ?";
    let count2: i64 = sqlx
    ::query_scalar(sql)
        .bind(jd)
        .bind(zbid)
        .fetch_one(get_my_pool()).await
        .map_err(|e| anyhow::anyhow!("获取kpi_import_log表中的记录数失败: {}", e))?;
    if count2 == 0 {
        let sql =
            "INSERT INTO pas_oltp_sh.kpi_import_log
                                (jd, zbid,confname)
                                SELECT ?,ZBID,IF(ISDF= 1,'kpi_import_pjzb',IF(ISDF=0,'kpi_import_lhzb',''))confname
                                from pas_oltp_sh.SH_CONF_ZB_DEPT
                                where ZBID = ? and  NF = LEFT(?, 4)";
        sqlx
        ::query(sql)
            .bind(jd)
            .bind(zbid)
            .bind(jd.to_string())
            .execute(get_my_pool()).await
            .map_err(|e| anyhow::anyhow!("插入kpi_import_log表失败: {}", e))?;
    }

    Ok(())
}

pub async fn install_kpi_export_log(jd: i32) -> anyhow::Result<()> {
    let sql = "SELECT COUNT(*)count  from pas_oltp_sh.kpi_import_log where jd = ?";
    let count: i64 = sqlx
    ::query_scalar(sql)
        .bind(jd)
        .fetch_one(get_my_pool()).await
        .map_err(|e| anyhow::anyhow!("获取kpi_import_log表中的记录数失败: {}", e))?;
    if count > 0 {
        return Ok(());
    }
    // 获取原有锁确保线程安全
    let lock = get_lock().await;
    let _guard = lock.lock().await;
    let count: i64 = sqlx
    ::query_scalar(sql)
        .bind(jd)
        .fetch_one(get_my_pool()).await
        .map_err(|e| anyhow::anyhow!("获取kpi_import_log表中的记录数失败: {}", e))?;
    if count > 0 {
        return Ok(());
    }
    let sql =
        "INSERT INTO pas_oltp_sh.kpi_import_log
                        (jd, zbid,confname)
                        SELECT ?,ZBID ,IF(ISDF= 1,'kpi_import_pjzb',IF(ISDF=0,'kpi_import_lhzb',''))confname
                        from pas_oltp_sh.SH_CONF_ZB_DEPT
                        where NF = LEFT(?, 4)";
    sqlx
    ::query(sql)
        .bind(jd)
        .bind(jd.to_string())
        .execute(get_my_pool()).await
        .map_err(|e| anyhow::anyhow!("插入kpi_import_log表失败: {}", e))?;

    Ok(())
}

pub async fn kpi_update_log(emid: u32, zbid: i32, jd: i32) -> anyhow::Result<()> {
    let drsj = chrono::Local::now().naive_local();
    let sql = "UPDATE pas_oltp_sh.kpi_import_log SET drsj = ?,emid = ? WHERE zbid = ? AND jd = ?";
    sqlx::query(sql)
        .bind(drsj)
        .bind(emid)
        .bind(zbid)
        .bind(jd)
        .execute(get_my_pool()).await?;
    Ok(())
}
